package com.gator.rabbitmq.pablo_rabbitmq.worker;


import com.gator.rabbitmq.pablo_rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

import java.io.IOException;

/**
 * @Author PABLO
 * @Date 2022/6/4 22:18
 * @Desc  多个消费者轮询消费同一个队列中的数据
 */
public class WorkerConsumer01 {


    public static final String QUEUE_NAME = "hello";

    public static void main(String[] args) {
        Channel channel = ConnectionUtils.getConnectionAndChannel();
        //接受信息
        //消费
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //在这里应答
            //getDeliveryTag交付标记，即哪个消息
            //multiple == false ，只应答当前tag，处理一个应答一个，保证消息不丢失
            System.out.println( "工作线程01" + new String(message.getBody()));
            channel.basicAck(message.getEnvelope().getDeliveryTag(),false);

        };
        //消费者取消消费
        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("msg被中断...");
        };
        //消费
        try {

            //修改为不公平分发,默认0,即能者多劳
           // channel.basicQos(1);
            System.out.println("01等待接受消息");
            //手动应答，autoAck设置为FALSE
            channel.basicConsume(QUEUE_NAME, false, deliverCallback, cancelCallback);


        } catch (IOException e) {
            e.printStackTrace();
        }

    }
}

class WorkerConsumer02 {


    public static final String QUEUE_NAME = "hello";

    public static void main(String[] args) {
        Channel channel = ConnectionUtils.getConnectionAndChannel();
        //接受信息
        //消费
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            try {
                //当执行到这里时停止当前线程，broker收不到ack，会将消息分发给consumer01处理
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println( "工作线程02" + new String(message.getBody()));
            //在这里应答
            //getDeliveryTag交付标记，即哪个消息
            //multiple == false ，只应答当前tag，处理一个应答一个，保证消息不丢失
            channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
        };
        //消费者取消消费
        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("msg被中断...");
        };
        //消费
        try {

            //修改为不公平分发,默认0
            //channel.basicQos(1);
            System.out.println("02等待接受消息");
            channel.basicConsume(QUEUE_NAME, false, deliverCallback, cancelCallback);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }
}
